package kafkahelper

import (
	"errors"
	"gitee.com/wanjimao/common-util/helper"
)

type KafkaCommon interface {
	AddToKafka(w *WriterHelper, roleData interface{}, allowErrNum int, currentErrNum int, errSlice []string) error
}
type defaultKafkaCommon struct {
}

// 初始化Kafka
func NewKafkaCommon() defaultKafkaCommon {
	return defaultKafkaCommon{}
}

// 递归写入kafka写入卡夫卡
// writeData 写入kafka的数据
// allowErrNum 允许请求失败次数
// currentErrNum 当前失败次数
// errSlice 错误信息
func (s defaultKafkaCommon) AddToKafka(w *WriterHelper, writeData interface{}, allowErrNum int, currentErrNum int, errSlice []string) error {
	err := w.Push(writeData)
	//推送失败
	if nil != err {
		errSlice = append(errSlice, err.Error())
		currentErrNum++
		if currentErrNum > allowErrNum {
			return errors.New(helper.InterfaceHelperObject.ToString(errSlice))
		}
		return s.AddToKafka(w, writeData, allowErrNum, currentErrNum, errSlice)
	}
	//推送成功直接
	return nil
}
